通用业务方案

限制接口访问权限

• 为了避免远程调用接口的信息过于敏感直接对外部暴露[比如查询用户密码密文]，可能需要对一些远程调用接口做权限控制，只允许特定的用户角色访问，让某个接口只限于系统内部使用，不被外部访问，常用的限制方式如下，以下方法可以单独使用，也可以组合使用，在实际应用中需要根据业务需求和安全策略来选择最合适的方法。

1. 使用Spring Security进行权限控制：

   • 配置Spring Security，为特定的接口添加权限控制，只允许特定的用户角色访问

   [配置实例]

   • 在这个例子中，/internal/** 路径下的接口只有具有"INTERNAL"角色的用户才能访问

   @Override
   protected void configure(HttpSecurity http) throws Exception {
       http
           .authorizeRequests()
           .antMatchers("/admin/**").hasRole("ADMIN")
           .antMatchers("/internal/**").hasRole("INTERNAL")
           .anyRequest().authenticated()
           .and()
           .httpBasic();
   }
   

2. 使用IP白名单：

   • 如果你希望接口只能被特定的IP地址访问，你可以在Spring Security配置中添加IP白名单。

   [配置实例]

   • 在这个例子中，只有来自IP地址192.168.1.100的请求才能访问/internal/**路径下的接口。

   xxxxxxxxxx
   @Override
   protected void configure(HttpSecurity http) throws Exception {
       http
           .authorizeRequests()
           .antMatchers("/internal/**").permitAll()
           .anyRequest().authenticated()
           .and()
           .ip()
           .access("hasIpAddress('192.168.1.100')")
           .and()
           .httpBasic();
   }
   

3. 使用Spring Cloud Gateway或Zuul作为API网关：

   • 在API网关层面控制访问权限，只允许内部服务访问特定的路由。

   [配置实例]

   • 在这个例子中，/internal/** 路径下的接口被配置在Spring Cloud Gateway中，并且可以进一步配置安全规则来限制访问。

   xxxxxxxxxx
   spring:
     cloud:
       gateway:
         routes:
           - id: internal-service
             uri: lb://internal-service
             predicates:
               - Path=/internal/**
             filters:
               - RewritePath=/internal/(?<segment>.*), /$\segment
             security:
               - oauth2ResourceServer()
   

4. 在服务层面进行控制：

   • 在服务代码中添加逻辑，检查请求的来源，例如通过特定的HTTP头或者请求参数来判断请求是否来自内部系统。

   [配置实例]

   • 在这个例子中，只有带有X-Internal-Request头且值为true的请求才能访问/internal/data接口

   xxxxxxxxxx
   @RestController
   public class InternalController {
       @GetMapping("/internal/data")
       public ResponseEntity<?> getInternalData(@RequestHeader("X-Internal-Request") String internalRequest) {
           if (!"true".equals(internalRequest)) {
               return ResponseEntity.status(HttpStatus.FORBIDDEN).build();
           }
           // 处理请求
       }
   }
   

5. 使用内部网络和防火墙规则：

   • 将你的服务部署在内部网络中，并通过防火墙规则限制外部访问。这是一种常见的做法，可以确保只有内部网络中的服务可以访问特定的接口。

定时调度任务

• 定时任务是常用的操作，比如每天的支付宝账单对账、每个月的财务汇总，系统闲时定时统计当前系统中重要数据的操作

• 常见的定时调度任务实现方法有：基于Java原生Timer、基于延迟消息队列、基于Quartz任务调度框架、基于xxl-job、基于netty时间轮实现的io.netty.util.HashedWheelTimer、SpringBoot的定时任务注解@Scheduled

基于Timer

• 基于JDK的java.util.Timer的定时任务实现

  • 🔎：从JDK1.3起就有一个java.util.Timer类来实现延时、定时执行任务的功能；Timer的优点是简单易用，但是致命缺点是所有的任务都是被同一个线程执行的，同一时间只能有一个任务执行，剩下任务都得等，执行过程中如果有一个任务发生延迟或者异常都会影响后续其他任务的执行，异常甚至会导致后续任务作废，适合任务数不确定的短时定时任务、定时任务需要单独控制的场景比如分布式锁的有效时间续期，早期的Redisson框架就是基于Timer做的定时锁过期时间重置，随着版本更新也换成了netty的的时间轮实现io.netty.util.HashedWheelTimer

1. java.util.Timer的使用

   • void--->timer.schedule(TimerTask task, long delay, long period)

     • 功能解析：定义在定时任务线程创建以后在delay毫秒以后开始间隔period时间执行一次的定时任务，task是任务对象，TimeTask是抽象类，需要实现抽象run方法；delay是相对于定时调度线程启动时的初始延迟时间，单位是毫秒，period是两次执行的间隔时间，单位是毫秒

     • 使用示例：

       xxxxxxxxxx
       /**
        * Thread[main,5,main]| 定时任务初始时间:172418119 1863
        * Thread[Timer-0,5,main]| 定时任务执行时间:17241811 96874
        * Thread[Timer-0,5,main]| 定时任务执行时间:17241812 06887
        * Thread[Timer-0,5,main]| 定时任务执行时间:17241812 16889
        * Thread[Timer-0,5,main]| 定时任务执行时间:17241812 26904
        * */
       public static void main(String[] args) {
           System.out.println(Thread.currentThread()+"| 定时任务初始时间:"+System.currentTimeMillis());
           new Timer().schedule(new TimerTask() {
               @Override
               public void run() {
                   System.out.println(Thread.currentThread()+"| 定时任务执行时间:"+System.currentTimeMillis());
               }
           },5000,10000);
       }
       

       • 示例含义：定时线程启动后初始延时5秒钟开始每隔十秒打印定时任务执行时间

     • 补充说明：

       • 🔎：TimerTask实现了Runnable接口，是一个抽象类，实例化对象需要实现其中的抽象run方法

   • void--->schedule(TimerTask task, long delay)

     • 功能解析：定义在定时任务线程创建以后在delay毫秒以后开始只执行一次的定时任务，task是任务对象，TimeTask是抽象类，需要实现抽象run方法；delay是相对于定时调度线程启动时的初始延迟时间，单位是毫秒，执行完一次定时任务线程就会销毁

     • 使用示例：

       ​x
       public class DistributedRedisLock implements Lock {
       
           private StringRedisTemplate redisTemplate;
       
           private String lockName;
       
           private String hSetField;
       
           private long expire = 30;
       
           public DistributedRedisLock(StringRedisTemplate redisTemplate, String lockName, String uuid) {
               this.redisTemplate = redisTemplate;
               this.lockName = lockName;
               this.hSetField = uuid + ":" + Thread.currentThread().getId();
           }
       
           @Override
           public void lock() {
               this.tryLock();
           }
       
           @Override
           public void lockInterruptibly() throws InterruptedException {
       
           }
       
           //无参tryLock方法是使用默认有效时间30s作为锁的有效时间
           @Override
           public boolean tryLock() {
               try {
                   return this.tryLock(-1L, TimeUnit.SECONDS);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
               return false;
           }
       
           /**
            * 加锁方法
            * @param time
            * @param unit
            * @return
            * @throws InterruptedException
            */
           @Override
           public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
               if (time != -1){
                   this.expire = unit.toSeconds(time);
               }
               String script = "if redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
                       "then " +
                       "   redis.call('hincrby', KEYS[1], ARGV[1], 1) " +
                       "   redis.call('expire', KEYS[1], ARGV[2]) " +
                       "   return 1 " +
                       "else " +
                       "   return 0 " +
                       "end";
               while (!this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), hSetField, String.valueOf(expire))){
                   Thread.sleep(50);
               }
               // 加锁成功，返回之前，开启定时器自动续期
               this.renewExpire();
               return true;
           }
       
           /**
            * 解锁方法
            */
           @Override
           public void unlock() {
               String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 0 " +
                       "then " +
                       "   return nil " +
                       "elseif redis.call('hincrby', KEYS[1], ARGV[1], -1) == 0 " +
                       "then " +
                       "   return redis.call('del', KEYS[1]) " +
                       "else " +
                       "   return 0 " +
                       "end";
               Long flag = this.redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), hSetField);
               if (flag == null){
                   throw new IllegalMonitorStateException("this lock doesn't belong to you!");
               }
           }
       
           @Override
           public Condition newCondition() {
               return null;
           }
       
           //上锁以后延迟1/3锁有效时间去定时执行续期脚本；注意execute方法会自动将lua的返回值1转成true，将返回值0返回false，注意nil也会转成false；如果需要区分nil需要使用Long类型的返回值，对应nil转成null；
           //定时任务采用延时执行一次任务，执行完以后线程就销毁，如果续期成功再次执行该任务，续期失败说明锁已经不是当前线程的锁了，不再进行续期操作；如果不这么写就需要去解锁成功以后取消定时任务，需要考虑各种意外情况导致定时线程无法被取消导致的死锁的情况，实现起来比较复杂，但是这种实现在分布式锁比较多的情况下会占用很多的线程资源
           //这里更改此前的getId方法是因为定时续期的execute方法也需要获取field字段，但是不能使用getId中的逻辑，因为定时任务需要开新的定时任务线程，但是field字段需要业务线程的线程id，因此这里只是改进将getId的逻辑放到了获取锁时的实例化锁对象的构造方法中，实际上锁重入只是key和Field字段相同，实际上锁重入并不是同一个DistributedRedisLock对象
           private void renewExpire(){
               String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
                       "then " +
                       "   return redis.call('expire', KEYS[1], ARGV[2]) " +
                       "else " +
                       "   return 0 " +
                       "end";
               new Timer().schedule(new TimerTask() {
                   @Override
                   public void run() {
                       if (redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), hSetField, String.valueOf(expire))) {
                           renewExpire();
                       }
                   }
               }, this.expire * 1000 / 3);
           }
       }
       

       • 示例含义：DistributedRedisLock是基于Redis+java.util.Timer+lua脚本来实现自动续期的可重入分布式锁实现，其中的renewExpire()方法就是使用Timer实现分布式锁自动续期的业务逻辑代码，这里面的又上一个续期任务执行成功唤醒下一个定时续期任务的思想值得借鉴，这样可以避免出现意外导致定时任务停不下来

     • 补充说明：

       • 🔎：该方法一般用在控制定时任务线程需要满足一定条件才能继续执行避免定时任务因为意外情况无法自动停止的场景，比如分布式锁的自动续期

基于延迟队列

1. 订单业务逻辑

   

   

2. 带回滚的锁库存逻辑

   • 数据库mall-sms中的表sms_stock_order_task记录着当前那个订单正在锁库存，在表sms_stock_order_task_detail表中记录着商品id、锁定库存的数量、锁定库存所在仓库id、订单任务号；即锁库存的时候先给数据库保存要锁定的库存记录，然后再锁定库存；只要锁定库存成功，库存相关的三张表因为本地事务都会成功保存锁库存记录；如果锁失败了数据库因为事务不会有锁库存记录，库存也不会锁定成功

   • 这样库存锁定表中存在的就是下单成功锁定库存的记录和下单失败但是锁定库存成功的记录，我们可以考虑使用一个定时任务，每隔一段时间就扫描一次数据库，检查一下哪些订单没有创建但是有锁定库存的记录，把这些锁定库存记录拿出来重新把库存补偿回滚一下，但是使用定时任务来定期扫描整个数据库表是很麻烦的一件事，我们通过引入延时消息队列来实现定时功能

   • 延时队列的原理是当库存锁定成功以后我们将库存锁定成功的消息发送给延时队列，但是在一定时间内消息被暂存在延时队列中不要往外发送，即锁定库存成功的消息暂存在延时队列中一段时间，在订单支付时间过期以后我们将该消息发送给解锁库存服务，解锁库存服务去检查订单是否被取消，如果订单根本没有创建或者因为订单未支付而被自动取消了就去数据库根据对应的锁库存记录将库存解锁，即锁定的库存在订单最大失效时间以后才固定进行解锁，通过延时消息队列来控制这个定时功能

   • 延时队列锁库存的业务流程

     • 1️⃣：创建订单锁定库存成功后，给主题交换器stock-event-exchange发送库存工作单消息，消息的路由键stock.locked，被主题交换器路由到队列stock.delay.queue中等待50分钟过期时间

     • 2️⃣：库存工作单在消息存活时间到期以后，队列将消息的路由键更改为stock.release，通过主题交换器stock-event-exchange将消息路由到队列stock.release.stock.queue，由该队列将消息转发给消费者库存服务

     • 3️⃣：消费者库存服务收到库存工作单消息，检查订单服务对应订单的状态，如果订单未被成功创建或者订单未支付就解锁被锁定的库存

     xxxxxxxxxx
     /**
      * @author Earl
      * @version 1.0.0
      * @描述 库存服务RabbitMQ配置
      * @创建日期 2024/12/02
      * @since 1.0.0
      */
     @Configuration
     public class StockRabbitMQConfig {
         /**
          * @return {@link MessageConverter }
          * @描述 给容器中注入一个使用Jackson将消息对象序列化为json对象的消息转换器
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/11/01
          * @since 1.0.0
          */
         @Bean
         public MessageConverter messageConverter(){
             return new Jackson2JsonMessageConverter();
         }
         /**
          * @return {@link Queue }
          * @描述 库存延迟队列延时队列
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/11/26
          * @since 1.0.0
          */
         @Bean
         public Queue stockDelayQueue(){
             Map<String, Object> arguments = new HashMap<>();
             arguments.put("x-dead-letter-exchange","stock-event-exchange");
             arguments.put("x-dead-letter-routing-key","stock.release");
             arguments.put("x-message-ttl",120000);
             return new Queue("stock.delay.queue", true, false, false, arguments);
         }
     
         /**
          * @return {@link Queue }
          * @描述 库存延迟队列路由队列
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/11/26
          * @since 1.0.0
          */
         @Bean
         public Queue stockReleaseStockQueue(){
             return new Queue("stock.release.stock.queue",true,false,false);
         }
     
         /**
          * @return {@link Exchange }
          * @描述 库存服务通用主题交换器
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/11/26
          * @since 1.0.0
          */
         @Bean
         public Exchange stockEventExchange(){
             return new TopicExchange("stock-event-exchange",true,false);
         }
     
         /**
          * @return {@link Binding }
          * @描述 延迟队列的延时队列stock.delay.queue和库存服务通用交换器stock-event-exchange的绑定关系
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/11/26
          * @since 1.0.0
          */
         @Bean
         public Binding orderCreateOrderBinding(){
             return new Binding("stock.delay.queue",
                     Binding.DestinationType.QUEUE,
                     "stock-event-exchange",
                     "stock.locked",
                     null);
         }
     
         /**
          * @return {@link Binding }
          * @描述 延迟队列的路由队列stock.release.stock.queue和库存服务通用交换器stock-event-exchange的绑定关系
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/11/26
          * @since 1.0.0
          */
         @Bean
         public Binding orderReleaseOrderBinding(){
             return new Binding("stock.release.stock.queue",
                     Binding.DestinationType.QUEUE,
                     "stock-event-exchange",
                     "stock.release.#",
                     null);
         }
     }
     

   • 解锁库存逻辑

     • 需要解锁库存的场景：

       • 创建订单成功，订单过期没有支付被系统自动取消或者订单被用户手动取消时需要解锁库存

       • 创建订单过程中，远程调用库存服务锁定库存成功，但是调用其他服务时出现异常导致创建订单整个业务回滚，之前成功锁定的库存就需要自动解锁来实现回滚，使用Seata分布式事务性能太差，不适合下单这种高并发场景；基于柔性事务的可靠消息加最终一致性的分布式事务方案，在保证分布式事务下的性能同时，允许一定时间内的软一致性并确保库存数据的最终一致性

     • 只要库存锁定成功就给RabbitMQ中对应的库存延迟队列发送库存工作单消息，使用RabbitTemplate.convertAndSend()发送锁定库存消息，同时锁定库存以前我们要保存库存工作单信息[对应表wms_ware_order_task，保存订单Id、订单号]，锁定存库成功以后要保存库存工作单详情[对应表wms_ware_order_task_detail，给该表添加bigint类型字段ware_id锁定库存所在仓库id；int类型的lock_status，其中1表示已锁定、2表示已解锁、3表示已扣减；注意MP更改了字段需要更改相应的Mapper文件中的resultMap标签；保存商品sku、商品数量、库存工作单id、锁定库存所在仓库id、默认锁定状态是已锁定1]，锁定库存前先保存库存工作单，保存库存工作单是为了追溯锁定库存信息

       • 给消息队列发送的消息实体类直接写在common包下，该消息对象StockLockedTo保存库存工作单id、该工作单下所有工作单详情id列表，老师这里发送消息的时机错了，所有商品都锁定成了才给消息队列发送消息，否则本地事务会自动回滚，老师是锁定一个商品就发送一条消息，如果事务回滚了发出去的消息就撤不回来了，而且老师这里发送的消息是全量的库存工作单详情数据

       [消息]

       xxxxxxxxxx
       /**
        * @author Earl
        * @version 1.0.0
        * @描述 库存被成功锁定消息
        * @创建日期 2024/12/02
        * @since 1.0.0
        */
       @Data
       public class StockLockedMessage {
           /**
            * 库存工作单
            */
           private WareOrderTaskEntity wareOrderTask;
           /**
            * 库存工作单详情列表
            */
           private List<WareOrderTaskDetailEntity> wareOrderTaskDetail;
       }
       

       [锁定库存保存库存工作单并发送消息]

       xxxxxxxxxx
       /**
        * @param lockStock
        * @return {@link List }<{@link LockStockResultTo }>
        * @描述 根据订单和订单项数据锁定库存
        * @author Earl
        * @version 1.0.0
        * @创建日期 2024/12/01
        * @since 1.0.0
        */
       @Override
       @Transactional
       public List<LockStockResultTo> lockStock(LockStockTo lockStock) throws RRException{
           //2. 锁定库存前创建库存工作单
           WareOrderTaskEntity wareOrderTask = new WareOrderTaskEntity();
           BeanUtils.copyProperties(lockStock,wareOrderTask);
           wareOrderTaskService.save(wareOrderTask);
           //1. 锁定库存
           List<LockStockTo.OrderItem> orderItems = lockStock.getOrderItems();
           List<Long> skuIds = orderItems.stream().map(LockStockTo.OrderItem::getSkuId).collect(Collectors.toList());
           Map<Long, LockStockTo.OrderItem> orderItemOfSku = orderItems.stream().collect(Collectors.toMap(LockStockTo.OrderItem::getSkuId, orderItem -> orderItem));
           //查询商品所在的全部仓库
           List<WareIdsOfSkuIdTo> wareIdsOfSkuIdTos=baseMapper.getWareBySkuIds(skuIds);
           if(wareIdsOfSkuIdTos.size()!=skuIds.size()){
               throw new RRException(StatusCode.NO_STOCK_EXCEPTION.getMsg(),
                       StatusCode.NO_STOCK_EXCEPTION.getCode());
           }
           ArrayList<LockStockResultTo> lockStockResults = new ArrayList<>();
           ArrayList<WareOrderTaskDetailEntity> wareOrderTaskDetails = new ArrayList<>();
           for (WareIdsOfSkuIdTo wareIdsOfSkuIdTo : wareIdsOfSkuIdTos) {
               Long skuId = wareIdsOfSkuIdTo.getSkuId();
               String[] wareIds = wareIdsOfSkuIdTo.getWareIds().split(",");
               Boolean skuLocked = false;
               for (String wareId : Arrays.asList(wareIds)) {
                   if(baseMapper.tryLockStock(wareId,skuId,orderItemOfSku.get(skuId).getSkuQuantity())==1){
                       //准备锁定库存的响应数据
                       LockStockResultTo lockStockResult = new LockStockResultTo();
                       skuLocked = true;
                       lockStockResult.setLocked(true);
                       lockStockResult.setLockQuantity(orderItemOfSku.get(skuId).getSkuQuantity());
                       lockStockResult.setWareId(Long.parseLong(wareId));
                       lockStockResult.setSkuId(skuId);
                       lockStockResults.add(lockStockResult);
                       //准备库存工作单详情
                       WareOrderTaskDetailEntity wareOrderTaskDetail = new WareOrderTaskDetailEntity();
                       wareOrderTaskDetail.setTaskId(wareOrderTask.getId());
                       wareOrderTaskDetail.setSkuId(skuId);
                       wareOrderTaskDetail.setSkuName(orderItemOfSku.get(skuId).getSkuName());
                       wareOrderTaskDetail.setSkuNum(orderItemOfSku.get(skuId).getSkuQuantity());
                       wareOrderTaskDetail.setWareId(Long.parseLong(wareId));
                       wareOrderTaskDetails.add(wareOrderTaskDetail);
                       break;
                   }
               }
               if(!skuLocked){
                   throw new RRException("商品"+skuId+StatusCode.NO_STOCK_EXCEPTION.getMsg(),
                           StatusCode.NO_STOCK_EXCEPTION.getCode());
               }
           }
           //for (WareOrderTaskDetailEntity wareOrderTaskDetail : wareOrderTaskDetails) {
           //    wareOrderTaskDetailService.save(wareOrderTaskDetail);
           //}
           wareOrderTaskDetailService.saveBatch(wareOrderTaskDetails);
           StockLockedMessage stockLockedMessage = new StockLockedMessage();
           stockLockedMessage.setWareOrderTask(wareOrderTask);
           stockLockedMessage.setWareOrderTaskDetail(wareOrderTaskDetails);
           rabbitTemplate.convertAndSend("stock-event-exchange","stock.locked",stockLockedMessage);
           return lockStockResults;
       }
       

     • 使用@RabbitHandler监听锁定库存消息队列，获取到消息对象，按以下情况执行解锁库存逻辑

       • 1️⃣：创建订单过程中，库存锁定成功，但是接下来创建订单出现问题，整个订单回滚，被锁定的库存需要自动解锁

         • 只要库存工作单详情存在，就说明库存锁定成功，此时我们就要看订单状态来判断是否需要解锁库存，如果订单都没有说明订单没有被成功创建，此时就要使用库存工作单详情来解锁库存

         • 如果有订单查看订单状态，如果订单已经被取消就解锁库存，只要订单没有被取消就不能解锁库存，订单被取消字段status等于4

           • 根据库存工作单的id去订单服务查询订单实体类，如果订单不存在或者订单的status字段为4就调用unLockStock方法解锁库存

       • 2️⃣：订单创建失败是由于库存锁定失败导致的

         • 库存工作单数据没有是库存本地事务整体回滚导致的，库存工作单记录不会创建，锁定库存操作也会全部自动回滚，这种情况无需解锁

       • 解锁库存需要知道商品的skuId、锁定库存所在仓库id、锁定库存数量、库存工作单详情id，解锁就是将原来的增加的锁定库存的字段再减掉UPDATE wms_ware_sku SET stock_locked = stock_locked + #{num} WHERE sku_id=#{skuId} AND ware_id = #{wareId}

         • 🚁：自动应答的消息队列，一旦在消息的消费过程中出现异常导致消息无法被正常消费，消息就丢失了，比如Feign远程调用网络闪断，或者远程服务的Feign调用必须携带用户的登录状态但是实际请求没有携带用户登录状态被远程服务拦截，抛出异常终止后续方法执行，此时消息就彻底丢失了

         • 📓：使用配置spring.rabbitmq.listener.simple.acknowledge-mode=manual开启消息接收手动应答，在订单解锁成功以后使用方法channel.basicAck(message.getMessageProperties().getDeliveryTag(),false)来做消息接收手动应答，解锁成功立马手动应答，无需解锁什么也不用做立马手动应答，只要在应答后发生异常也不会导致消息丢失；如果远程调用没有成功返回在库存解锁以前出现问题，我们使用方法channel.basicReject(message.getMessageProperties().getDeliveryTag(),true)来手动拒绝消息并将消息重新放到队列中，给别人继续消费消息解锁库存的机会，比如由于分区故障为了保证一致性部分服务不可用

       • ❓：订单服务所有远程调用请求都要求有登录状态，但是我们的消息队列监听方法的远程调用不可能带用户登录状态，因此我们需要在订单服务的拦截器中放行所有消息队列监听方法调用的订单服务远程接口，特别注意，这个需要被放行的请求路径还拼接了请求参数导致请求路径是动态的

         • 🔑：我们通过在拦截器中放行指定URI的请求来实现这个目的，对于URI是变化的我们使用Spring提供的boolean match = AntPathMatcher.match("/order/order/status/**",request.getRequestURI())[HttpServletRequest.getRequestURI()是获取请求路径的URI，HttpServletRequest.getRequestURL()是获取请求路径的URL]，如果请求URI匹配我们需要的格式就直接通过拦截器的return true放行，无需再进行用户登录状态检查

         xxxxxxxxxx
         /**
          * @author Earl
          * @version 1.0.0
          * @描述 用户登录状态拦截器
          * @创建日期 2024/11/09
          * @since 1.0.0
          */
         public class LoginStatusInterceptor implements HandlerInterceptor {
         
             public static ThreadLocal<UserBaseInfoVo> loginUser=new ThreadLocal<>();
         
             @Override
             public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
                 AntPathMatcher antPathMatcher = new AntPathMatcher();
                 if(antPathMatcher.match("/order/order/get/**", request.getRequestURI())){
                     return true;
                 }
                 UserBaseInfoVo attribute = (UserBaseInfoVo)request.getSession().getAttribute(MallConstant.SESSION_USER_LOGIN_STATUS_KEY);
                 if(attribute!=null){
                     loginUser.set(attribute);
                     return true;
                 }else{
                     request.getSession().setAttribute("tip","请先登录");
                     response.sendRedirect("http://auth.earlmall.com/login.html");
                     return false;
                 }
             }
         
         }
         

       • 专门抽取一个消息队列的监听器Service来处理消息队列中的消息

         • 在类上标注@RabbitListener(queues="stock.release.stock.queue")来监听指定队列，在类上标注@Service注解将该类的实例化对象作为容器组件，在具体的方法上标注注解@RabbitHandler，在该方法中调用库存服务实现的解锁库存逻辑，解锁库存的方法出现任何异常都手动拒绝消息并重新入队列，只要解锁库存方法成功调用就手动应答接收消息，远程调用如果状态码不是0说明没有查到对应订单的实体类，此时直接抛异常执行拒绝接收消息的逻辑

         [监听队列消息]

         xxxxxxxxxx
         /**
          * @author Earl
          * @version 1.0.0
          * @描述 库存释放队列监听器
          * @创建日期 2024/12/02
          * @since 1.0.0
          */
         @Service
         @RabbitListener(queues = "stock.release.stock.queue")
         public class StockReleaseQueueListenerImpl extends ServiceImpl<MessageDao, MessageEntity> implements StockReleaseQueueListener {
             @Autowired
             private WareSkuService wareSkuService;
         
             /**
              * @param message
              * @param stockLockedMessage
              * @param channel
              * @描述 根据到期的锁定库存消息释放库存
              * @author Earl
              * @version 1.0.0
              * @创建日期 2024/12/02
              * @since 1.0.0
              */
             @RabbitHandler
             public void tryReleaseStock(Message message, StockLockedMessage stockLockedMessage, Channel channel) {
                 try{
                     wareSkuService.tryReleaseStock(stockLockedMessage.getWareOrderTask().getOrderSn());
                     channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                 }catch (Exception e){
                     try {
                         channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
                     } catch (Exception exception) {
                         MessageEntity messageLog = new MessageEntity();
                         messageLog.setMessageStatus(2);
                         messageLog.setClassType(stockLockedMessage.getClass().getTypeName());
                         messageLog.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
                         messageLog.setToExchange(message.getMessageProperties().getReceivedExchange());
                         messageLog.setContent(JSON.toJSONString(stockLockedMessage));
                         save(messageLog);
                     }
                 }
             }
         }
         

         [解锁库存]

         xxxxxxxxxx
         /**
          * @param orderSn
          * @描述 根据库存锁定消息尝试解锁库存
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/12/02
          * @since 1.0.0
          */
         @Override
         @Transactional
         public void tryReleaseStock(String orderSn) {
             //1. 根据订单号远程查询订单服务对应的订单状态，如果订单不存在或者订单状态不是已关闭就不解锁库存
             R res = orderFeignClient.getOrderByOrderSn(orderSn);
             OrderTo order = res.get("order", new TypeReference<OrderTo>() {});
             if(order==null && order.getStatus()==4){
                 return;
             }
             //2. 根据订单号查询库存工作单id
             WareOrderTaskEntity wareOrderTaskEntity = wareOrderTaskService.getOne(new QueryWrapper<WareOrderTaskEntity>().
                     eq("order_sn", orderSn));
             if(wareOrderTaskEntity==null){
                 return;
             }
             List<WareOrderTaskDetailEntity> wareOrderTaskDetails = wareOrderTaskDetailService.list(new QueryWrapper<WareOrderTaskDetailEntity>().
                     eq("task_id", wareOrderTaskEntity.getId()));
             for (WareOrderTaskDetailEntity wareOrderTaskDetail : wareOrderTaskDetails) {
                 //检索库存工作单状态，只有库存工作单状态为未解锁时才解锁库存，如果不为未解锁本条库存不解锁
                 if(wareOrderTaskDetail.getLockStatus()!=2){
                     baseMapper.releaseStock(wareOrderTaskDetail.getSkuNum(),wareOrderTaskDetail.getSkuId(),wareOrderTaskDetail.getWareId());
                     //解锁后将库存工作单状态更改为已解锁
                     wareOrderTaskDetail.setLockStatus(2);
                     wareOrderTaskDetailService.updateById(wareOrderTaskDetail);
                 }
             }
         }
         

       • 解锁库存成功后通过库存工作单详情id将库存工作单详情的状态lock_status更改为已解锁2，增加前面解锁库存的条件只有库存工作单详情为已锁定状态且需要解锁时才能解锁库存

     • 带回滚的锁库存实现

       • 给库存服务mall-ware引入、配置RabbitMQ并在主启动类上标注@EnableRabbit开启RabbitMQ功能

       • 配置RabbitMQ的消息的JSON序列化机制

       • 给库存服务添加一个默认交换器stock-event-exchange

         • 交换器使用Topic交换器类型，因为该交换器需要绑定多个队列，而且还需要使用对不同消息的路由键进行模糊匹配的功能

       • 给库存服务添加一个释放库存队列stock.release.stock.queue，支持持久化，不支持排他和自动删除，普通队列不需要设置参数

       • 给库存服务添加一个延迟库存工作单消息的队列stock.delay.queue，给该延时队列设置死信交换器stock-event-exchange，设置死信的路由键为stock.release，设置队列的消息存活时间为120秒[方便测试用的，比验证订单创建的延迟队列多一分钟]，支持持久化，不支持排他和自动删除

       • 给库存服务的库存释放队列和交换器添加一个绑定关系，绑定目的地stock.release.stock.queue，交换器stock-event-exchange，绑定键stock.release.#

       • 给库存服务的库存延时队列和交换器添加一个绑定关系，绑定目的地stock.delay.queue，交换器stock-event-exchange，绑定键stock.delay.#

       • 监听一个队列让以上所有容器组件都通过SpringBoot自动去RabbitMQ中检查创建

       • 取消订单逻辑

         • 订单服务队列和交换器组件和绑定关系

           xxxxxxxxxx
           /**
            * @author Earl
            * @version 1.0.0
            * @描述 自定义RabbitMQ配置
            * 1. 使用@Bean注解注入容器的队列、交换器、绑定关系如果在RabbitMQ服务器中没有SpringBoot会自动在RabbitMQ服务器中进行创建
            * @创建日期 2024/11/26
            * @since 1.0.0
            */
           @Configuration
           public class CustomRabbitMQConfig {
               /**
                * @return {@link Queue }
                * @描述 订单延迟队列延时队列
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/26
                * @since 1.0.0
                */
               @Bean
               public Queue orderDelayQueue(){
                   Map<String, Object> arguments = new HashMap<>();
                   arguments.put("x-dead-letter-exchange","order-event-exchange");
                   arguments.put("x-dead-letter-routing-key","order.release.order");
                   arguments.put("x-message-ttl",60000);
                   return new Queue("order.delay.queue", true, false, false, arguments);
               }
           
               /**
                * @return {@link Queue }
                * @描述 订单延迟队列路由队列
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/26
                * @since 1.0.0
                */
               @Bean
               public Queue orderReleaseOrderQueue(){
                   return new Queue("order.release.order.queue",true,false,false);
               }
           
               /**
                * @return {@link Exchange }
                * @描述 订单服务通用主题交换器
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/26
                * @since 1.0.0
                */
               @Bean
               public Exchange orderEventExchange(){
                   return new TopicExchange("order-event-exchange",true,false);
               }
           
               /**
                * @return {@link Binding }
                * @描述 延迟队列的延时队列order.delay.queue和订单服务通用交换器order-event-exchange的绑定关系
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/26
                * @since 1.0.0
                */
               @Bean
               public Binding orderCreateOrderBinding(){
                   return new Binding("order.delay.queue",
                           Binding.DestinationType.QUEUE,
                           "order-event-exchange",
                           "order.create.order",
                           null);
               }
           
               /**
                * @return {@link Binding }
                * @描述 延迟队列的路由队列order.release.order.queue和订单服务通用交换器order-event-exchange的绑定关系
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/26
                * @since 1.0.0
                */
               @Bean
               public Binding orderReleaseOrderBinding(){
                   return new Binding("order.release.order.queue",
                           Binding.DestinationType.QUEUE,
                           "order-event-exchange",
                           "order.release.order",
                           null);
               }
           
               @Bean
               public Binding orderReleaseStockBinding(){
                   return new Binding("stock.release.stock.queue",
                           Binding.DestinationType.QUEUE,
                           "order-event-exchange",
                           "order.release.other.#",
                           null);
               }
           }
           
           

         • 订单创建成功就给交换器order-event-exchange发送消息，消息路由键order.create.order，保存的消息是OrderCreateTO.getOrder()，消息会被交换器路由到order.delay.queue[队列中的消息延时时间为30min]，消息变成死信后路由键配置成order.release.order并将消息转发到order-event-exchange路由到order.release.order.queue被订单服务监听，订单服务监听接收取消订单并将消息转发以路由键order.release.other通过交换器order-event-exchange，队列将消息转发给订单服务

           [订单服务创建订单发起消息]

           xxxxxxxxxx
           /**
            * @param orderParam
            * @return {@link PayVo }
            * @描述 创建订单
            * @author Earl
            * @version 1.0.0
            * @创建日期 2024/11/30
            * @since 1.0.0
            */
           @Override
           //@GlobalTransactional
           @Transactional
           public PayVo createOrder(OrderSubmitParamVo orderParam) throws RRException{
               UserBaseInfoVo userBaseInfo = LoginStatusInterceptor.loginUser.get();
               String orderSn = IdWorker.getTimeId();
               Long userId = userBaseInfo.getId();
               //1. 校验订单提交令牌，校验失败抛出表单重复提交异常
               if(!verifyOrderSubmitToken(userId,orderParam.getToken())){
                   throw new RRException(StatusCode.FORM_REPEAT_EXCEPTION.getMsg(),
                           StatusCode.FORM_REPEAT_EXCEPTION.getCode());
               }
               //2. 创建订单
               OrderEntity order = createOrder(userId, orderParam.getAddrId(),orderSn);
               //3. 创建订单项列表
               //远程调用购物车服务通过用户id获取购物车被选中的购物项
               List<OrderStatementVo.SelectedCartItemVo> selectedCartItems = cartFeignClient.getSelectedCartItem(userId);
               if(selectedCartItems==null || selectedCartItems.size()<=0){
                   throw new RRException(StatusCode.NO_CART_ITEM_EXCEPTION.getMsg(),
                           StatusCode.NO_CART_ITEM_EXCEPTION.getCode());
               }
               List<OrderItemEntity> orderItems = createOrderItems(orderSn,selectedCartItems);
               //4. 计算订单价格
               calculateOrderAmount(order,orderItems);
               //7. 验价
               if (Math.abs(order.getPayAmount().subtract(orderParam.getTotalPrice()).doubleValue())>=0.01) {
                   throw new RRException(StatusCode.PRICE_VERIFY_EXCEPTION.getMsg()+
                           "核算金额: ￥"+orderParam.getTotalPrice()+",实际金额: ￥"+order.getPayAmount(),
                   StatusCode.PRICE_VERIFY_EXCEPTION.getCode());
               }
               //5. 保存订单记录和订单项记录
               orderService.save(order);
               //seata目前在AT模式下不支持批量插入记录，https://blog.csdn.net/qq_33240556/article/details/140790581,反正我们后面要换成软性事务，后面再换成批量插入
               //for (OrderItemEntity orderItem : orderItems) {
               //    orderItemService.save(orderItem);
               //}
               orderItemService.saveBatch(orderItems);
               //6. 锁定库存
               LockStockTo lockStock = new LockStockTo();
               lockStock.setOrderId(order.getId());
               lockStock.setOrderSn(orderSn);
               lockStock.setConsignee(order.getReceiverName());
               lockStock.setConsigneeTel(order.getReceiverPhone());
               lockStock.setDeliveryAddress(order.getReceiverDetailAddress());
               lockStock.setPaymentWay(1);
               lockStock.setOrderItems(orderItems);
               R res = stockFeignClient.lockStock(lockStock);
               if (res.getCode()!=0) {
                   throw new RRException((String) res.get("msg"),res.getCode());
               }
               // 8.给消息队列发送消息
               OrderCreatedMessage orderCreatedMessage = new OrderCreatedMessage();
               orderCreatedMessage.setOrderSn(orderSn);
               rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",orderCreatedMessage);
               //封装视图数据，注意存入分布式session的数据对应实体类最好全部放在common包下，session中有的数据不论其他服务能否用到，
               // 其他服务必须包含session中所有数据的对应类型，否则其他服务即使没有使用对应的session也会直接报错
               PayVo pay = new PayVo();
               OrderVo orderVo = new OrderVo();
               BeanUtils.copyProperties(order,orderVo);
               List<OrderItemVo> orderItemVos = orderItems.stream().map(orderItem -> {
                   OrderItemVo orderItemVo = new OrderItemVo();
                   BeanUtils.copyProperties(orderItem, orderItemVo);
                   return orderItemVo;
               }).collect(Collectors.toList());
               pay.setOrder(orderVo);
               pay.setOrderItems(orderItemVos);
               pay.setFare(order.getFreightAmount());
               pay.setPayablePrice(order.getPayAmount());
               return pay;
           }
           

           [订单服务接收消息取消订单并向库存服务发送消息]

           xxxxxxxxxx
           /**
            * @author Earl
            * @version 1.0.0
            * @描述 库存释放队列监听器
            * @创建日期 2024/12/02
            * @since 1.0.0
            */
           @Service
           @RabbitListener(queues = "order.release.order.queue")
           public class OrderReleaseQueueListenerImpl extends ServiceImpl<MessageDao,MessageEntity> implements OrderReleaseQueueListener {
               @Autowired
               RabbitTemplate rabbitTemplate;
               @Autowired
               OrderService orderService;
           
               /**
                * @param message
                * @param orderCreatedMessage
                * @param channel
                * @描述 根据到期的锁定库存消息释放库存
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/12/02
                * @since 1.0.0
                */
               @RabbitHandler
               public void trySendMessageReleaseStock(Message message, OrderCreatedMessage orderCreatedMessage, Channel channel) {
                   orderService.cancelOrder(orderCreatedMessage.getOrderSn());
                   try{
                       rabbitTemplate.convertAndSend("order-event-exchange","order.release.other",orderCreatedMessage);
                       channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                   }catch (Exception e){
                       try {
                           channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
                       } catch (Exception exception) {
                           try {
                               channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                           } catch (IOException ioException) {
                               ioException.printStackTrace();
                           }
                           MessageEntity messageLog = new MessageEntity();
                           messageLog.setMessageStatus(2);
                           messageLog.setClassType(orderCreatedMessage.getClass().getTypeName());
                           messageLog.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
                           messageLog.setToExchange(message.getMessageProperties().getReceivedExchange());
                           messageLog.setContent(JSON.toJSONString(orderCreatedMessage));
                           save(messageLog);
                       }
                   }
               }
           }
           

         • 订单服务收到消息根据订单id查询数据库对应的订单状态，如果订单状态为订单创建对应的状态码，将订单状态更改为取消订单对应状态码

           xxxxxxxxxx
           /**
            * @param orderSn
            * @描述 根据订单号关闭订单
            * @author Earl
            * @version 1.0.0
            * @创建日期 2024/12/03
            * @since 1.0.0
            */
           @Override
           public void cancelOrder(String orderSn) {
               OrderEntity order = getOne(new QueryWrapper<OrderEntity>().eq("order_sn", orderSn));
               order.setStatus(OrderConstant.OrderStatus.CANCELLED.getCode());
               updateById(order);
           }
           

         • 通过监听消息队列消息在取消订单或者订单创建失败的情况下解锁库存

           [消息队列监听]

           xxxxxxxxxx
           /**
            * @author Earl
            * @version 1.0.0
            * @描述 库存释放队列监听器
            * @创建日期 2024/12/02
            * @since 1.0.0
            */
           @Service
           @RabbitListener(queues = "stock.release.stock.queue")
           public class StockReleaseQueueListenerImpl extends ServiceImpl<MessageDao, MessageEntity> implements StockReleaseQueueListener {
               @Autowired
               private WareSkuService wareSkuService;
           
               /**
                * @param message
                * @param stockLockedMessage
                * @param channel
                * @描述 根据到期的锁定库存消息释放库存
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/12/02
                * @since 1.0.0
                */
               @RabbitHandler
               public void tryReleaseStock(Message message, StockLockedMessage stockLockedMessage, Channel channel) {
                   String orderSn = stockLockedMessage.getWareOrderTask().getOrderSn();
                   tryReleaseStock(orderSn,message,channel,JSON.toJSONString(stockLockedMessage),stockLockedMessage.getClass().toString());
               }
           
               /**
                * @param message
                * @param orderCreatedMessage
                * @param channel
                * @描述 取消订单自动解锁库存
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/12/03
                * @since 1.0.0
                */
               @RabbitHandler
               public void tryReleaseStock(Message message, OrderCreatedMessage orderCreatedMessage,Channel channel){
                   String orderSn = orderCreatedMessage.getOrderSn();
                   tryReleaseStock(orderSn,message,channel,JSON.toJSONString(orderCreatedMessage),orderCreatedMessage.getClass().toString());
               }
           
               /**
                * @param orderSn
                * @param message
                * @param channel
                * @param messageContent
                * @param messageClassType
                * @描述 解锁库存方法
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/12/03
                * @since 1.0.0
                */
               private void tryReleaseStock(String orderSn,Message message,Channel channel,String messageContent,String messageClassType){
                   try{
                       wareSkuService.tryReleaseStock(orderSn);
                       channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                   }catch (Exception e){
                       try {
                           channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
                       } catch (Exception exception) {
                           try {
                               MessageEntity messageLog = new MessageEntity();
                               messageLog.setMessageStatus(2);
                               messageLog.setClassType(messageClassType);
                               messageLog.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
                               messageLog.setToExchange(message.getMessageProperties().getReceivedExchange());
                               messageLog.setContent(JSON.toJSONString(messageContent));
                               save(messageLog);
                               channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                           } catch (Exception excep) {
                               excep.printStackTrace();
                           }
                       }
                   }
               }
           }
           

           [解锁库存]

           xxxxxxxxxx
           /**
            * @param orderSn
            * @描述 根据库存锁定消息尝试解锁库存
            * @author Earl
            * @version 1.0.0
            * @创建日期 2024/12/02
            * @since 1.0.0
            */
           @Override
           @Transactional
           public void tryReleaseStock(String orderSn) {
               //1. 根据订单号远程查询订单服务对应的订单状态，如果订单不存在或者订单状态不是已关闭就不解锁库存
               R res = orderFeignClient.getOrderByOrderSn(orderSn);
               OrderTo order = res.get("order", new TypeReference<OrderTo>() {});
               if(order==null && order.getStatus()==4){
                   return;
               }
               //2. 根据订单号查询库存工作单id
               WareOrderTaskEntity wareOrderTaskEntity = wareOrderTaskService.getOne(new QueryWrapper<WareOrderTaskEntity>().
                       eq("order_sn", orderSn));
               if(wareOrderTaskEntity==null){
                   return;
               }
               List<WareOrderTaskDetailEntity> wareOrderTaskDetails = wareOrderTaskDetailService.list(new QueryWrapper<WareOrderTaskDetailEntity>().
                       eq("task_id", wareOrderTaskEntity.getId()));
               for (WareOrderTaskDetailEntity wareOrderTaskDetail : wareOrderTaskDetails) {
                   //检索库存工作单状态，只有库存工作单状态为未解锁时才解锁库存，如果不为未解锁本条库存不解锁
                   if(wareOrderTaskDetail.getLockStatus()!=2){
                       baseMapper.releaseStock(wareOrderTaskDetail.getSkuNum(),wareOrderTaskDetail.getSkuId(),wareOrderTaskDetail.getWareId());
                       //解锁后将库存工作单状态更改为已解锁
                       wareOrderTaskDetail.setLockStatus(2);
                       wareOrderTaskDetailService.updateById(wareOrderTaskDetail);
                   }
               }
           }
           

         • ❓：我们这里是用库存解锁时间大于取消订单时间来实现解锁库存只要订单的状态为已取消或者订单没有成功创建，就释放已经锁定的库存，但是这种方式存在很严重的问题；比如订单创建成功，但是由于各种原因，消息延迟了很久才发给消息队列，但是库存一锁定成功就将消息发送给消息队列了，导致解锁库存的消息比取消订单的消息先到期，这时候就会导致解锁库存的消息被消费，库存因为订单处于新建状态无法解锁，即使后续订单被解锁了库存也无法被解锁了；即一旦发生意外导致解锁库存的消息比取消订单的消息先到，就会发生被锁定的库存永远无法解锁的情况

           • 🔑：让订单服务取消订单后再发一个消息路由键为order.release.other给交换器order-event-exchange，我们为交换器order-event-exchange和队列stock.release.stock.queue设定绑定关系，绑定关系设定为order.release.other.#，让取消订单的消息被队列stock.release.stock.queue发送给消费者库存服务。库存服务用@RabbitListener(queues="stock.release.stock.queue")监听同一个队列stock.release.stock.queue，用@RabbitHandler标注的方法监听消息类型为OrderTo，在原来解锁库存的逻辑中判断，当前库存是否解锁过，没解锁过就解锁，解锁过就不用解锁了，老师的逻辑是根据订单号查询库存工作单，根据库存工作单找到所有没有解锁的库存工作单详情调用此前解锁库存的方法进行解锁，感觉这里老师的实现不好，自己实现这部分代码

       • 实际上解锁库存是订单取消的时候解锁一次，锁定库存成功以后一定时间再解锁一次

      

消息可靠性投递

1. 影响消息可靠性的因素

   • 消息丢失：消息丢失在电商系统中是一个非常可怕的操作，比如订单消息丢失可能会影响到后续一连串比如商家确认、解锁库存、物流等等各种信息，消息可能发生丢失的原因如下：

     • 消息从生产者发送出去，但是由于网络问题抵达RabbitMQ服务器失败，或者因为异常根本没有发送成功

       • 这时候可以用try...catch语句块来发送消息，发送失败在catch语句块中设置重试策略

       • 同时给数据库创建一张消息数据库表mq_message，建表语句如下

         xxxxxxxxxx
         CREATE TABLE `mq_message` (
             `message_id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '消息id',
             `content` text COMMENT '消息内容',#序列化为json
             `to_exchange` varchar(255) DEFAULT NULL COMMENT '投递交换器',
             `routing_key` varchar(255) DEFAULT NULL COMMENT '路由键',
             `class_type` varchar(255) DEFAULT NULL COMMENT '消息类型的全限定类名',
             `message_status` int(1) DEFAULT '0' COMMENT '0-新建 1-已发送 2-错误抵达 3-已抵达',
             `create_time` datetime DEFAULT NULL COMMENT '消息日志创建时间',
             `update_time` datetime DEFAULT NULL COMMENT '消息日志更新时间',
             PRIMARY KEY (`message_id`)
         ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
         

       • 只要消息发送失败就给数据库存上这么一条日志，定期扫描数据库来检查消息日志状态来重新发送消息

     • 消息到达Broker，消息只有被投递给队列才算持久化完成，一旦消息还没有到达队列，RabbitMQ服务器宕机消息就会因为还没有来得及持久化而发生丢失

       • 开启生产者消息抵达队列确认，只要消息没有成功抵达队列就会触发生产者的returnCallback回调，消息不能成功抵达应该设置消息重试发送和向数据库记录消息日志

       • 开启生产者消息确认回调，只要消息成功抵达RabbitMQ服务器就触发该回调

     • 自动ACK的状态，消费者收到消息，但是消息没有被成功消费，比如消费消息或者消费消息前出现异常或者服务器宕机，自动应答的消息会直接丢失

       • 开启手动ACK，消息成功消费以后再手动应答接收消息，消息消费失败就手动拒绝消息让消息重新入队列，注意消息没有被应答即没有手动拒绝RabbitMQ没有收到应答的消息也会默认重新入队列再次发送

     • 🔎：防消息丢失的核心就是做好消息生产者和消息消费者两端的消息确认机制，主要策略就是生产者的消息抵达确认回调和消费者的手动应答，凡是消息不能成功抵达服务端和消费端的消息都做好消息日志记录，定期扫描数据库，将发送失败的消息定期重新发送

   • 消息重复：就是因为各种原因导致的消息重新投递

     • 消息消费成功，事务已经提交，但是手动Ack的时候机器宕机或者网络连接中断导致手动Ack没有进行，RabbitMQ的消息因为没有收到应答自动将消息重新入队列并将消息状态从Unack状态变成ready状态，并再次将消息发送给消费者

     • 消息消费过程中消费失败又再次重试发送消息，注意啊，虽然我们让消息消费失败消息拒绝重新入队列

       • 解决办法是业务消息消费接口设计成幂等性接口，比如解锁库存要判断库存工作单详情的状态位，消息消费成功修改对应状态位

       • 使用redis或者mysql防重表，将消息和业务通过唯一标识联系起来，业务被成功处理过的消息就不用再处理了

       • RabbitMQ的每个消息都有一个redelivered消息属性字段，每个消息都可以通过Boolean redelivered = message.getMessageProperties().getRedelivered()判断当前消息是否被第二次或者第N次重新投递过来的，这个一般做辅助判断，因为谁也不能保证消息在第几次消费被消费成功

   • 消息积压：消息队列中的消息积压太多，导致消息队列的性能下降

     • 消费者宕机导致消息积压

     • 消费者消费能力不足，比如活动高峰期，比如消费者宕机导致的消费者集群消费能力不足，有服务完全不可用消息反复重入队列消息肯定会积压，应该设置重试次数，投递达到重试次数消息就被专门的服务处理比如存入数据库离线处理

       • 注意消费者没有应答消费消息，队列中的消息处于Unack状态，生产者会不停报错，让CPU飚高，非常消耗系统性能，这个问题要想办法防一下

     • 发送者发送消息的流量太大，超出消费者的消费能力

       • 限制发送者的流量，让服务限流业务进不来就能限制发送者的流量，不过只是因为消息中间件或者消费者能力有限就限制业务有点得不偿失

       • 上线更多的消费者增强消息的消费能力

       • 上线专门的消息队列消息消费服务，将消息批量从消息队列中取出来，直接写入数据库，缓解消息队列压力，然后再缓慢离线从数据库中获取消息离线处理

       • 消息队列集群

2. 一般都是把消息中间件专门做成一个服务，叫数据中台，负责消息发送和自动记录消息日志，消息发送失败自动进行重试，将消息发送的所有功能都考虑周到，其他服务通过调用该服务来实现消息的发送，看老师的意思，一般消息发送成功也得记录日志，这个可以作为防止消息丢失更进一步的手段，毕竟会影响性能

3. 生产者抵达确认带数据库保存失败消息

   xxxxxxxxxx
   /**
    * @author Earl
    * @version 1.0.0
    * @描述 RabbitMQ客户端自定义配置
    * @创建日期 2024/11/01
    * @since 1.0.0
    */
   @Slf4j
   @Configuration
   public class MallRabbitConfig {
   
       @Autowired
       private RabbitTemplate rabbitTemplate;
   
       /**
        * @return {@link MessageConverter }
        * @描述 给容器中注入一个使用Jackson将消息对象序列化为json对象的消息转换器
        * @author Earl
        * @version 1.0.0
        * @创建日期 2024/11/01
        * @since 1.0.0
        */
       @Bean
       public MessageConverter messageConverter(){
           return new Jackson2JsonMessageConverter();
       }
   
       @PostConstruct
       public void initRabbitTemplate(){
           /*
           1. 设置RabbitMQ服务器收到消息后的确认回调ConfirmCallback
               配置配置项spring.rabbitmq.publisher-confirms=true
               为rabbitTemplate设置回调实例化对象confirmCallback
           */
           rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
               /**
                * @param correlationData 当前消息的唯一关联标识，里面的id就是消息标识的唯一id
                * @param ack 消息是否成功收到
                * @param cause 消息发送失败的原因
                * @描述 1. 只要消息抵达Broker就会触发该回调，与消费者和消息是否入队列无关
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/02
                * @since 1.0.0
                */
               @Override
               public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                   log.info("message confirm: {correlationData:"+correlationData+
                           "ack:"+ack+
                           "cause:"+cause+"}");
               }
           });
           /*
           2.设置RabbitMQ队列没有收到消息的确认回调ReturnCallback
                 配置配置项spring.rabbitmq.publisher-returns=true
                 配置配置项spring.rabbitmq.template.mandatory=true
                 为rabbitTemplate设置回调实例化对象returnCallback
           * */
           rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
               /**
                * @param message 投递失败的消息本身的详细信息
                * @param replyCode 导致消息投递失败的错误状态码
                * @param replyText 导致消息投递失败的错误原因
                * @param exchange 当时该消息发往的具体交换器
                * @param routingKey 当时该消息的具体路由键
                * @描述
                * @author Earl
                * @version 1.0.0
                * @创建日期 2024/11/03
                * @since 1.0.0
                */
               @Override
               public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                   log.info("message lose: {message:"+message+
                           "replyCode:"+replyCode+
                           "replyText:"+replyText+
                           "exchange:"+exchange+
                           "routingKey"+routingKey);
               }
           });
   
       }
   }
   

4. 消费者手动ACK

   xxxxxxxxxx
   /**
    * @author Earl
    * @version 1.0.0
    * @描述 库存释放队列监听器
    * @创建日期 2024/12/02
    * @since 1.0.0
    */
   @Service
   @RabbitListener(queues = "stock.release.stock.queue")
   public class StockReleaseQueueListenerImpl extends ServiceImpl<MessageDao, MessageEntity> implements StockReleaseQueueListener {
       @Autowired
       private WareSkuService wareSkuService;
   
       /**
        * @param message
        * @param stockLockedMessage
        * @param channel
        * @描述 根据到期的锁定库存消息释放库存
        * @author Earl
        * @version 1.0.0
        * @创建日期 2024/12/02
        * @since 1.0.0
        */
       @RabbitHandler
       public void tryReleaseStock(Message message, StockLockedMessage stockLockedMessage, Channel channel) {
           String orderSn = stockLockedMessage.getWareOrderTask().getOrderSn();
           tryReleaseStock(orderSn,message,channel,JSON.toJSONString(stockLockedMessage),stockLockedMessage.getClass().toString());
       }
   
       /**
        * @param message
        * @param orderCreatedMessage
        * @param channel
        * @描述 取消订单自动解锁库存
        * @author Earl
        * @version 1.0.0
        * @创建日期 2024/12/03
        * @since 1.0.0
        */
       @RabbitHandler
       public void tryReleaseStock(Message message, OrderCreatedMessage orderCreatedMessage,Channel channel){
           String orderSn = orderCreatedMessage.getOrderSn();
           tryReleaseStock(orderSn,message,channel,JSON.toJSONString(orderCreatedMessage),orderCreatedMessage.getClass().toString());
       }
   
       /**
        * @param orderSn
        * @param message
        * @param channel
        * @param messageContent
        * @param messageClassType
        * @描述 解锁库存方法
        * @author Earl
        * @version 1.0.0
        * @创建日期 2024/12/03
        * @since 1.0.0
        */
       private void tryReleaseStock(String orderSn,Message message,Channel channel,String messageContent,String messageClassType){
           try{
               wareSkuService.tryReleaseStock(orderSn);
               channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
           }catch (Exception e){
               try {
                   channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
               } catch (Exception exception) {
                   try {
                       MessageEntity messageLog = new MessageEntity();
                       messageLog.setMessageStatus(2);
                       messageLog.setClassType(messageClassType);
                       messageLog.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
                       messageLog.setToExchange(message.getMessageProperties().getReceivedExchange());
                       messageLog.setContent(JSON.toJSONString(messageContent));
                       save(messageLog);
                       channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                   } catch (Exception excep) {
                       excep.printStackTrace();
                   }
               }
           }
       }
   }
   

5. 使用MP数据库消息日志记录

   • 同时给数据库创建一张消息数据库表mq_message，建表语句如下

     xxxxxxxxxx
     CREATE TABLE `mq_message` (
         `message_id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '消息id',
         `content` text COMMENT '消息内容',#序列化为json
         `to_exchange` varchar(255) DEFAULT NULL COMMENT '投递交换器',
         `routing_key` varchar(255) DEFAULT NULL COMMENT '路由键',
         `class_type` varchar(255) DEFAULT NULL COMMENT '消息类型的全限定类名',
         `message_status` int(1) DEFAULT '0' COMMENT '0-新建 1-已发送 2-错误抵达 3-已抵达',
         `create_time` datetime DEFAULT NULL COMMENT '消息日志创建时间',
         `update_time` datetime DEFAULT NULL COMMENT '消息日志更新时间',
         PRIMARY KEY (`message_id`)
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
     

   • MP插入消息记录

     [实体类]

     xxxxxxxxxx
     /**
      * @author Earl
      * @version 1.0.0
      * @描述 库存服务消息实体
      * @创建日期 2024/12/02
      * @since 1.0.0
      */
     @Data
     @TableName("sms_message")
     public class MessageEntity implements Serializable {
         private static final long serialVersionUID = 1L;
         /**
          * 消息id
          */
         @TableId
         private Long messageId;
         /**
          * 消息内容
          */
         private String content;
         /**
          * 投递交换器
          */
         private String toExchange;
         /**
          * 路由键
          */
         private String routingKey;
         /**
          * 消息类型的全限定类名
          */
         private String classType;
         /**
          * 0-新建 1-已发送 2-错误抵达 3-已抵达
          */
         private Integer messageStatus;
         /**
          * 消息日志创建时间
          */
         @TableField(fill = FieldFill.INSERT)
         private Date createTime;
         /**
          * 消息日志更新时间
          */
         @TableField(fill = FieldFill.INSERT_UPDATE)
         private Date updateTime;
     }
     

     [持久化接口]

     xxxxxxxxxx
     /**
      * @author Earl
      * @version 1.0.0
      * @描述 库存服务消息
      * @创建日期 2024/12/02
      * @since 1.0.0
      */
     @Mapper
     public interface MessageDao extends BaseMapper<MessageEntity> {
     }
     

     [持久化接口对应xml]

     xxxxxxxxxx
     <?xml version="1.0" encoding="UTF-8"?>
     <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
     
     <mapper namespace="com.earl.mall.stock.dao.MessageDao">
     
         <!-- 可根据自己的需求，是否要使用 -->
         <resultMap type="com.earl.mall.stock.entity.MessageEntity" id="messageMap">
             <result property="messageId" column="message_id"/>
             <result property="content" column="content"/>
             <result property="toExchange" column="to_exchange"/>
             <result property="routingKey" column="routing_key"/>
             <result property="classType" column="class_type"/>
             <result property="messageStatus" column="message_status"/>
             <result property="createTime" column="create_time"/>
             <result property="updateTime" column="update_time"/>
         </resultMap>
     
     </mapper>
     

     [业务实现类]

     xxxxxxxxxx
     /**
      * @author Earl
      * @version 1.0.0
      * @描述 库存释放队列监听器
      * @创建日期 2024/12/02
      * @since 1.0.0
      */
     @Service
     @RabbitListener(queues = "stock.release.stock.queue")
     public class StockReleaseQueueListenerImpl extends ServiceImpl<MessageDao, MessageEntity> implements StockReleaseQueueListener {
         @Autowired
         private WareSkuService wareSkuService;
     
         /**
          * @param message
          * @param stockLockedMessage
          * @param channel
          * @描述 根据到期的锁定库存消息释放库存
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/12/02
          * @since 1.0.0
          */
         @RabbitHandler
         public void tryReleaseStock(Message message, StockLockedMessage stockLockedMessage, Channel channel) {
             String orderSn = stockLockedMessage.getWareOrderTask().getOrderSn();
             tryReleaseStock(orderSn,message,channel,JSON.toJSONString(stockLockedMessage),stockLockedMessage.getClass().toString());
         }
     
         /**
          * @param message
          * @param orderCreatedMessage
          * @param channel
          * @描述 取消订单自动解锁库存
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/12/03
          * @since 1.0.0
          */
         @RabbitHandler
         public void tryReleaseStock(Message message, OrderCreatedMessage orderCreatedMessage,Channel channel){
             String orderSn = orderCreatedMessage.getOrderSn();
             tryReleaseStock(orderSn,message,channel,JSON.toJSONString(orderCreatedMessage),orderCreatedMessage.getClass().toString());
         }
     
         /**
          * @param orderSn
          * @param message
          * @param channel
          * @param messageContent
          * @param messageClassType
          * @描述 解锁库存方法
          * @author Earl
          * @version 1.0.0
          * @创建日期 2024/12/03
          * @since 1.0.0
          */
         private void tryReleaseStock(String orderSn,Message message,Channel channel,String messageContent,String messageClassType){
             try{
                 wareSkuService.tryReleaseStock(orderSn);
                 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
             }catch (Exception e){
                 try {
                     channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
                 } catch (Exception exception) {
                     try {
                         MessageEntity messageLog = new MessageEntity();
                         messageLog.setMessageStatus(2);
                         messageLog.setClassType(messageClassType);
                         messageLog.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
                         messageLog.setToExchange(message.getMessageProperties().getReceivedExchange());
                         messageLog.setContent(JSON.toJSONString(messageContent));
                         save(messageLog);
                         channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                     } catch (Exception excep) {
                         excep.printStackTrace();
                     }
                 }
             }
         }
     }
     

      

基于Quartz

• 这是非常流行的任务调度器框架，官网：Quartz Enterprise Job Scheduler

1. 定时任务的执行时刻

单点登录

• 对于多生态的超大系统，比如尚硅谷[atguigu.com]旗下有尚硅谷在线教育[gulixueyuan.com]、有尚硅谷电子商城[gulimall.com]、众筹系统[gulifunding.com]等等，如果我们每个小系统都设置一个账号，用户使用旗下的每个产品都需要注册一个账号，这样用户使用体验非常不好

• 在这种超大型系统中，我们希望专门抽取一个认证中心专门来处理用户登录业务，只要用户在一个小系统中进行了登录，任何其他系统都能保持用户的登录状态，在任何一个小系统中登出，就在整个系统中的其他应用也登出，而且这些小系统的顶级域名还可以不一样，这就是单点登录

• 这种单点登录是不能使用session来解决不同系统中用户登录状态的识别问题，因为下发cookie的作用域最大只能放大到一级域名即顶级域名，只要请求访问的顶级域名变化了就无法携带其他顶级域名和子域服务下发的cookie，就没法拿着这个cookie去验证用户的登录状态

开源项目

1. 开源的单点登录demo

   • 码云搜索徐雪里/xxl-sso，这是一个XXL社区提供的分布式单点登录框架，下载压缩包或者克隆到本地

   • 项目目录结构

     • xxl-sso-core是核心包

     • xxl-sso-server是登录中心服务器

     • xxl-sso-samples是一些简单的例子，有基于cookie和session的xxl-sso-web-sample-springboot，也有基于token的xxl-sso-token-sample-springboot，这两个就是下面使用逻辑的客户端

     

   • 认证中心配置文件解析[配置文件在目录xxl-sso-server中，后面没有特殊说明都在该目录下]

     xxxxxxxxxx
     ### web
     server.port=8080 #服务器的端口是8080
     server.servlet.context-path=/xxl-sso-server #访问路径为8080/xxl-sso-server
     
     ### resources
     spring.mvc.servlet.load-on-startup=0
     spring.mvc.static-path-pattern=/static/**
     spring.resources.static-locations=classpath:/static/
     
     ### freemarker
     spring.freemarker.templateLoaderPath=classpath:/templates/
     spring.freemarker.suffix=.ftl
     spring.freemarker.charset=UTF-8
     spring.freemarker.request-context-attribute=request
     spring.freemarker.settings.number_format=0.##########
     
     ### xxl-sso
     xxl.sso.redis.address=redis://127.0.0.1:6379 #配置了redis，我们需要将其改成自己的redis服务器配置
     xxl.sso.redis.expire.minute=1440
     

   • 测试目录xxl-sso-samples下客户端的配置文件解析

     [xxl-sso-web-sample-springboot]

     • 使用命令mvn clean package -Dmaven.skip.test=true，如果单独对这个子项目打包，注意这个子项目依赖核心包xxl-sso-core，在对xxl-sso-web-sample-springboot进行打包的时候会去本地仓库找对应的核心包依赖，只打包当前项目是不会安装到本地仓库的，因此需要先在核心包xxl-sso-core的pom.xml所在目录下使用命令mvn install将该核心包安装到本地仓库中，这种自定义的所有被依赖包必须在本地仓库有了才能对当前项目进行打包

       • 我们可以直接在父项目pom.xml所在使用命令mvn clean package -Dmaven.skip.test=true一次性将所有的子项目都打包好，这个就是谷粒商城的手动打包方式，如果是整体打包，打包子项目就不需要单独将被依赖的包安装到本地仓库中

     • 使用命令java -jar xxl-sso-web-sample-springboot-1.1.1-SNAPSHOT.jar --server.port=8081来启动测试实例项目

       • 注意可以在配置文件更改服务的端口，也可以在启动命令中更改服务实例的端口

     • 使用命令java -jar xxl-sso-web-sample-springboot-1.1.1-SNAPSHOT.jar --server.port=8082来启动测试实例项目

     xxxxxxxxxx
     ### web
     server.port=8081
     server.servlet.context-path=/xxl-sso-web-sample-springboot #访问路径默认是8081/xxl-sso-web-sample-springboot
     
     ### resources
     spring.mvc.servlet.load-on-startup=0
     spring.mvc.static-path-pattern=/static/**
     spring.resources.static-locations=classpath:/static/
     
     ### freemarker
     spring.freemarker.templateLoaderPath=classpath:/templates/
     spring.freemarker.suffix=.ftl
     spring.freemarker.charset=UTF-8
     spring.freemarker.request-context-attribute=request
     spring.freemarker.settings.number_format=0.##########
     
     ### xxl-sso
     xxl.sso.server=http://xxlssoserver.com:8080/xxl-sso-server #这个是认证中心的地址，这个地址需要和实际认证中心的地址保持一致
     xxl.sso.logout.path=/logout
     xxl-sso.excluded.paths=
     xxl.sso.redis.address=redis://127.0.0.1:6379 #配置redis将redis改成我们自己的服务器配置
     

2. 框架业务使用示例和逻辑

   • 我们准备三个顶级域名ssoserver.com、client1.com、client2.com，我们将ssoserver作为登录认证服务器，将client1.com和client2.com作为我们的小系统，用来模拟跨顶级域名的单点登录测试效果，小系统就是上面说的xxl-sso-samples目录下

   • 使用SwitchHost更改三个域名都映射到本机，如果不使用SwitchHost需要更改路径C:\windows\System32\drivers\etc\hosts

   • 启动服务xxl-sso-server，我们在启动的时候需要将源文件打包，pom.xml中直接聚合了三个目录，我们可以一次性将整个项目都打包了，在pom.xml所在目录打开cmd窗口，使用命令mvn clean package -Dmaven.skip.test=true清包打包跳过Maven测试[即使没有打过包的也要进行清包，其他项目也是如此，避免出现莫名其妙的问题]，打完包直接使用java -jar命令启动

     xxxxxxxxxx
     <modules>
         <module>xxl-sso-core</module>
         <module>xxl-sso-server</module>
         <module>xxl-sso-samples</module>
     </modules>
     

   • 通过请求路径http://ssoserver.com:8080/xxl-sso-server访问认证中心

   • 分别使用两个端口启动项目xxl-sso-web-sample-springboot，注意把配置文件给改对了，要保证redis服务器的地址和认证中心的地址正确

   • 分别通过请求路径http://client1.com:8081/xxl-sso-web-sample-springboot和http://client2.com:8082/xxl-sso-web-sample-springboot访问一个测试服务两个服务实例

   • 验证任意一个认证中心登录，所有不同顶级域名的服务实例都登录，任意一个服务实例登出，所有不同顶级域名的服务实例都登出

接口幂等性

1. 方案一：Token令牌机制

   • 最常见的场景是验证码场景，比如买票锁定座位12306需要我们输入验证码，请求中只有验证码正确的情况下才能锁定座位成功，而且验证码使用一次就失效

   • 应用在提交订单业务上我们可以在给用户响应订单确认页的同时下发一个Token令牌，服务器存储了该令牌，用户提交订单的时候携带该令牌，只要提交订单的请求携带了该令牌，我们就验证通过创建订单，只要一次验证令牌通过服务器就删除该令牌，用户的多次提交最终只有一个请求能验证通过

   • 如何比较完美地执行验证令牌操作才能尽可能保证不出错，主要考虑是先删令牌再创建订单还是先创建订单再删除令牌，

     • 如果是后删令牌问题非常大，前一个请求处于创建订单期间第二个请求就开始执行令牌验证操作，由于前一个请求订单还没创建令牌还没来得及删除，第二个请求会验证通过，此时第二个请求也会开始创建订单，因此我们一般首选先删令牌

     • 如果是先删令牌

       • 可能存在订单没有成功创建服务器宕机导致订单创建失败的问题；

       • 同时分布式场景下服务器的令牌不会直接存在本地缓存，一般都是存在redis中，从Redis中存取数据就会存在延迟，此时如果两个请求间隔时间很短，两个请求都成功从Redis中获取到令牌，同时验证成功，也会发生接口的幂等性问题；因此我们一般会要求验证令牌时获取令牌、验证令牌和删除令牌三个操作是一个原子性操作，我们可以使用lua脚本if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end来保证三步操作的原子性，让验证操作直接在Redis中原子性一步执行

2. 方案二：锁机制

   • 数据库悲观锁

     • select * from xxx where id=1 for update;来使用数据库行级锁，悲观锁一般结合事务一起使用，定位记录的查询条件一定要设置成主键或者唯一键索引，否则很容易造成锁表，锁表处理起来非常麻烦

   • 数据库乐观锁

     • 数据库乐观锁update t_goods set count=count-1,version=version+1 where goodId=2 and version=1;更适合更新场景，给数据库表添加一个版本字段，只有获取的数据版本和更新时的数据版本相同更新操作才会成功，更新操作会单增数据版本，只要一次数据成功更新，多次请求中后续到达的请求就无法再修改数据了[这种场景很适合库存数据的更新，比如第一次操作库存服务更新了库存数据并更新了版本号，但是操作库存数据的请求响应给订单服务时出现了网络问题，订单服务再次调用库存服务，但是订单服务最初获取的库存数据版本是旧的，库存数据的版本已经发生了更新，后续的重试不论执行多少遍都不会成功，我们可以使用额外信息来判断更新操作的具体订单，乐观锁适合读多写少的业务场景]

   • 分布式锁

     • 多台服务实例执行相同的写操作，我们对被操作对象上分布式锁，拿到锁的机器执行写操作，写的同时给数据一个标志位，当其他机器拿到锁以后先检查标志位发现数据已经发生了写操作就直接释放锁不再执行写操作了

3. 方案三：唯一约束

   • 唯一键索引：比如给用户的订单确认页下发唯一标识并保留在服务器内部，给该唯一标识对应的字段添加唯一索引，生成订单向数据库插入该唯一标识，只有第一次生成的订单数据因为唯一键索引能成功写入数据库，后续插入订单记录因为唯一键约束就会插入失败

   • Redis的set防重：数据只能被处理一次，我们可以在处理数据时计算数据的MD5并将MD5密文存入Redis的set中，每次处理数据前先检查一下数据的MD5是否已经存在，如果已经存在就不处理了，百度网盘的秒传功能就是这样的

4. 方案四：防重表

   • 上面的Redis防重也算防重表的一种

   • 比如建立一个去重表，使用订单号orderNo作为去重表的唯一键索引，将订单号插入去重表再做业务处理，保证订单号插入去重表和业务处理在同一个事务中；这样因为去重表中有订单号作为唯一键索引，多次提交的后续请求就会因为无法向去重表插入数据导致请求失败，业务处理不会执行；同时即使去重表数据插入成功，只要业务处理失败因为事务也会将去重表的数据回滚，让订单的后续重试能够继续进行

     • 因为要保证去重表和业务表的整体事务，需要将去重表和业务表放在同一个数据库中

5. 方案五：给请求下发全局唯一标识ID

   • 调用接口时我们可以给请求指定一个全局唯一的ID，接口处理该请求的时候将全局唯一的请求ID存储到Redis中，处理业务请求的时候如果发现该请求标识已经存在了我们就不再对该请求进行处理直接返回成功

   • 这种请求全局唯一标识ID还可以做服务调用链路追踪，追踪请求经过了哪些服务

   • Nginx给每个请求分配唯一Id的配置proxy_set_header X-Request-Id $request_id;，这个一般做链路追踪，不能做防重处理，因为Nginx给每个请求分配的全局唯一ID都是不一样的，即重复提交的每个请求Nginx都会给请求分配一个全局的唯一ID；我们自定义的请求可以参考这种思路给请求头设置一个全局唯一的请求标识ID，特别是做Feign的远程调用请求，重试的请求请求头中的唯一标识ID都是一样的

Token令牌方案示例

• 我们采用Token令牌机制来给订单确认页返回一个唯一令牌，并将该令牌存入Redis中，感觉存到session中最好，这样可以天然避免第三方恶意操作用户的订单token令牌

1. 前端业务逻辑

   • 我们给订单确认页返回一个Token令牌，这里为了简单直接使用UUID，用一个type属性值为hidden的input框以token作为输入框的值来保存该Token令牌，这样能直接在表单提交时自动作为参数上传

   • 给分布式session中即Redis中存放该Token令牌，这里为了避免发生恶意对Token令牌乱用，同时session自动就有有效时间，用户不关浏览器但是一段时间不操作Token令牌就会自动失效，用户没有确认订单关闭了浏览器Token令牌也会直接失效，老师是使用session中存储的用户id加token的名字的方式作为key存储的，value存储token本身

     xxxxxxxxxx
     //7. 设置用户提交订单的校验令牌
     String orderSubmitToken = UUID.randomUUID().toString().replace("-", "");
     redisTemplate.opsForValue().set(OrderConstant.ORDER_SUBMIT_TOKEN_PREFIX+":"+userBaseInfo.getId(),
             orderSubmitToken,
             OrderConstant.ORDER_SUBMIT_TOKEN_TTL,
             TimeUnit.MINUTES);
     orderStatement.setToken(orderSubmitToken);
     

   • 封装提交订单时上传的数据[为了防止前端数据上传恶意篡改导致的数据错误引起问题纠纷]，给提交订单按钮做一个form表单，所有要提交的数据都做成隐藏的input框，实际上该项目只提交了收货地址、订单最后价格、防重令牌

     • 封装用户收货地址的ID，页面初始化或者用户切换收货地址的时候将输入框的值进行回填$("#addrIdInput").val(addrId)

     • 封装用户的支付方式Integer

     • 商品无需提交，再去购物车获取一遍商品[京东也是这样做的，这样可以避免确认订单请求被恶意篡改，就是我们先调出订单确认页，此时已经计算了购物车中的选中商品价格，我们再修改购物车被选中商品，此时总价肯定发生了变化，我们在原来的订单确认页点击提交订单，我们发现最新的价格变成了购物车选中商品后的价格，这说明两个情况，第一是京东提交订单以后是重新获取购物车数据来计算最终价格的，第二是京东没有对订单确认页的商品金额做验价]，重新计算金额

     • 防重令牌，做接口幂等性校验，页面模板引擎渲染时赋值

     • 封装订单确认页总价，提交订单后重新获取购物车商品数据计算价格以后对价格进行校验，如果校验通过说明购物车中被选中的商品在订单确认期间没有发生变化，如果校验没通过说明购物车中的商品发生了变化，我们可以通知用户购物车商品发生了变化，让用户注意一下，计算应付总价时赋值

     • 订单备注信息，这个这里没实现，有需要再实现

     xxxxxxxxxx
     <form method="post" action="/order/order/submit">
         <input type="hidden" name="addrId" id="addr-id">
         <input type="hidden" name="totalPrice" id="total-price">
         <input type="hidden" name="token" th:value="${orderStatement.token}">
         <button type="submit" class="tijiao">提交订单</button>
     </form>
     <script>
         function queryFare(){
             $.get("http://earlmall.com/api/ware/wareordertask/calculate/fare?addrId="+$(".addr-receive p[selectedStatus='1']").attr("addrId"),function (data){
                 var userAddress = data.fare.userAddress;
                 $(".yfze_b").text("寄送至： "+
                     userAddress.province+" "+
                     userAddress.city+" "+
                     userAddress.region+" "+
                     userAddress.detailAddress+" 收货人："+
                     userAddress.name+" "+
                     userAddress.phone);
                 $("#addr-id").val(userAddress.id);
                 calculateFareAndPayablePrice(data.fare.fare);
             })
         }
         function calculateFareAndPayablePrice(fare){
             $("#fare").text("￥ "+fare*1+".00");
             $(".hq").text("￥ "+([[${orderStatement.payablePrice}]]*1+fare*1)+".00");
             $("#total-price").val([[${orderStatement.payablePrice}]]*1+fare*1);
         }
     </script>
     

2. 后端创建订单逻辑

   • 视图跳转逻辑

     • 下单成功跳转支付页

     • 下单失败携带异常信息跳回订单确认页重新确认订单

     xxxxxxxxxx
     /**
      * @param orderParam
      * @return {@link String }
      * @描述 创建用户订单
      * @author Earl
      * @version 1.0.0
      * @创建日期 2024/11/30
      * @since 1.0.0
      */
     @PostMapping("/order/submit")
     public String createOrder(OrderSubmitParamVo orderParam, HttpSession session, RedirectAttributes attributes){
         try{
             PayVo pay = orderWebService.createOrder(orderParam);
             session.setAttribute("pay",pay);
             return "redirect:http://order.earlmall.com/pay.html";
         }catch (RRException e){
             attributes.addFlashAttribute("error",e.getMsg());
             return "redirect:http://order.earlmall.com/statement.html";
         }
     }
     
     @GetMapping("/pay.html")
     public String getViewPage(){
         return "pay";
     }
     

   • 验证令牌，创建用户订单，校验价格，锁订单库存

     xxxxxxxxxx
     /**
      * @param orderParam
      * @return {@link PayVo }
      * @描述 创建订单
      * @author Earl
      * @version 1.0.0
      * @创建日期 2024/11/30
      * @since 1.0.0
      */
     @Override
     @GlobalTransactional
     @Transactional
     public PayVo createOrder(OrderSubmitParamVo orderParam) throws RRException{
         UserBaseInfoVo userBaseInfo = LoginStatusInterceptor.loginUser.get();
         String orderSn = IdWorker.getTimeId();
         Long userId = userBaseInfo.getId();
         //1. 校验订单提交令牌，校验失败抛出表单重复提交异常
         if(!verifyOrderSubmitToken(userId,orderParam.getToken())){
             throw new RRException(StatusCode.FORM_REPEAT_EXCEPTION.getMsg(),
                     StatusCode.FORM_REPEAT_EXCEPTION.getCode());
         }
         ...
     }
     /**
      * @param token
      * @return {@link Long }
      * @描述 校验并删除提交订单的令牌
      * @author Earl
      * @version 1.0.0
      * @创建日期 2024/11/30
      * @since 1.0.0
      */
     private boolean verifyOrderSubmitToken(Long userId,String token){
         String key = OrderConstant.ORDER_SUBMIT_TOKEN_PREFIX + ":" + userId;
         String script = "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
         return redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                 Arrays.asList(key), token)==1;
     }
     

      

本地事务

1. 本地事务

   • 数据库事务的特性[ACID]：

     • 原子性[一系列操作整体性不可拆分，即要么整体成功、要么整体失败]、

     • 一致性[整体数据操作前后守恒]、

     • 隔离性[或独立性，事务之间相互隔离，一个业务操作失败回滚不会影响其他业务操作]、

     • 持久性[事务一旦成功提交，数据就一定会落盘到数据库，认为是先落盘再提示事务成功提交]

   • 本地事务的应用场景是单体应用连接一个数据库，没有多个数据库、没有涉及服务拆分、也没有涉及服务间的远程调用

2. Spring提供的本地事务注解@Transactional注解

   • Spring框架提供了一个@Transactional注解来使用本地事务

   • 隔离级别：隔离级别是SQL数据库规定的一些规范

     • READ UNCOMMITTED[读未提交]：设置该隔离级别的事务可以读到其他未提交事务的数据，这会导致脏读现象[比如读未提交的数据被回滚了，但是其他操作已经拿到回滚前的数据进行后续的计算]

     • READ COMMITTED[读已提交]：设置该隔离级别的事务可以读取已经提交事务的数据，这是Oracle和SqlServer的隔离级别，特点是每次读取都读取的是实际值，同一系列操作中两次读取的数据不同都是实际值

     • REPEATABLE READ[可重复读]：设置该隔离级别的事务读取到的数据是事务开始时的数据，事务期间读取的数据都是相同的，这是MYSQL默认的隔离级别，特点是存在幻读现象，即实际数据在业务处理期间已经发生变化，但仍然使用的事务开启时的数据

       • mysql的InnoDB引擎可以通过next-key locks即行锁算法机制来避免幻读

     • SERIALIZABLE[序列化]：设置该隔离级别的事务全是串行顺序执行的，MySQL数据库的InnoDB引擎会给读操作隐式加一把读共享锁，避免脏读、不可重复读、幻读问题，但是这也意味着使用这种隔离级别数据库操作就失去了并发能力

     • 🔎：通过@Transactional(isolation=Isolation.READ_COMMITTED)可以指定当前事务的隔离级别

   • 事务传播行为[Spring中的事务管理行为]：事务的传播行为就是指被调用方法的事务和调用方法的事务之间的关系，比如a方法需要开启一个事务，b方法也需要开启一个事务，a方法调用b方法，那么a，b方法开启的事务之间的关系

     • PROPAGATION_REQUIRED：如果当前业务操作还没有事务，就创建一个新事务；如果当前业务操作已经存在事务就加入该事务，一旦事务中任何一处失败事务中的所有操作都回滚

       • 一旦b方法的事务设置成PROPAGATION_REQUIRED，如果b方法被开启了事务的a方法调用，b方法的事务配置就会完全失效，比如b方法单独设置了代码执行超过7s就会回滚@Transactional(propagation=Propagation.REQUIRED,timeout=7)，a方法设置了代码执行超过30s就会回滚，b方法被a方法调用，那么b方法的事务配置就会直接失效，即a事务的设置会自动传播到和a方法共用一个事务的方法

     • PROPAGATION_REQUIRED_NEW：不论当前业务操作有没有事务，都为被@Transactional注解标注的方法创建一个新的事务，该方法出现问题只会回滚该方法的业务操作，其他事务出问题不会影响该方法的事务行为

     • PROPAGATION_SUPPORTS：如果当前业务操作已经存在事务就加入该事务，如果当前业务操作还没有事务就以非事务的方式执行

     • PROPAGATION_MANDATORY：如果当前业务操作已经存在事务就加入该事务，如果当前业务操作还没有事务就直接抛出异常

     • PROPAGATION_NOT_SUPPORTED：以非事务的方式执行当前被标注方法，如果当前业务操作已经存在事务，就将当前事务挂起

     • PROPAGATION_NEVER：以非事务的方式执行当前被标注方法，如果当前业务操作已经存在事务，就直接抛出异常

     • PROPAGATION_NESTED：如果当前业务操作已经存在事务，被标注方法会在当前事务内部创建一个子事务，子事务的特点是会在子事务开启的时刻创建一个保存点，子事务失败只会回退到该保存点，不会回滚整个事务，父事务中的其他操作仍然能正常执行；但是如果父事务回滚，子事务也会一起回滚；如果当前业务操作还没有事务就为被标注方法创建一个全新的事务

       • 这个事务传播行为和PROPAGATION_REQUIRED_NEW很像，被标注方法的事务失败不会影响业务操作的其他事务，主要区别如下；PROPAGATION_REQUIRES_NEW会挂起当前事务，并创建一个全新的事务，这意味着新事务与原事务完全独立，事务成功或者失败不会相互影响。而PROPAGATION_NESTED则是在当前事务内部创建一个子事务，如果父事务被回滚，子事务也会被回滚，但子事务的回滚不会影响到父事务。

3. SpringBoot中的本地事务

   • SpringBoot中也是默认使用Spring的本地事务注解@Transactional

   • SpringBoot中使用@Transactional注解的坑

     • ❓：在SpringBoot中，如果a方法、b方法、c方法是同一个Service中的方法，三个方法都标注了@Transactional注解且都做了个性化配置，比如都各自配置了不同的事务传播行为和超时时间，此时如果a方法调用了b方法和c方法，b方法和c方法的任何事物配置都不会生效，包括事物传播行为，都是和a方法共用同一个事务

     • 🔑：​这是AOP的问题，在之前SpringCache中就出现过，那儿老师没有解释，事务注解是通过AOP实现的，事务是通过代理对象orderService来控制的，如果直接调用同一个类中的实例方法本质上相当于跳过代理对象直接通过方法名调用同一个类中的方法，就类似于代码的复制粘贴，被调用的方法事务注解不会生效；根本原因就是绕过了代理对象，这一块不太熟，后面复习的时候深入理解一下，a方法是通过代理对象调用的，但是b方法和c方法只是单纯地将代码复制粘贴过来，我们通过orderService来调用被标注了基于AOP实现的事务或者缓存注解的方法对应的注解会生效，但是同一个类下的代码的相互调用是类似于直接复制拷贝代码的形式，通过this来调用也是没用的，this也会被处理成同一个对象从而直接复制拷贝被调用方法的代码，只有通过代理对象类似于orderService.a()来调用事务注解或者缓存注解才会生效，总之基于AOP实现的注解都需要通过代理对象来调用才会生效，通过this调用也是不生效的；而且一定不要企图在orderService中自动注入orderService来通过代理对象调用b方法或者c方法，这相当于orderService依赖于orderService，在构造orderService将其注入容器时会发现构造orderService自身有一个属性需要注入orderService，会造成循环依赖的问题，系统启动的时候就要爆炸

     • 🔑：同一个对象内基于AOP实现的注解标注方法相互调用注解功能失效的解决办法，核心是要使用代理对象来调用标注了基于AOP实现的注解的方法

       • 引入Spring的AOP动态代理场景启动器spring-boot-starter-aop，引入该场景启动器的目的是使用其依赖的aspectjweaver，这个动态代理更加强大

       • 在配置类上使用注解@EnableAspectJAutoProxy开启Aspectj动态代理，不使用该注解默认使用的是JDK默认的按照接口自动生成的动态代理，使用该注解所有的动态代理都是Aspectj创建的，使用Aspectj的好处是即使没有接口也可以创建动态代理；在@EnableAspectJAutoProxy中指定exposeProxy属性为true即@EnableAspectJAutoProxy(exposeProxy=true)来对外暴露代理对象

       • 只要设置了通过Aspectj创建代理对象，我们就可以在任何地方通过org.springframework.aop.framework.AopContext即AOP上下文的AopContext.currentProxy()拿到当前代码所在对象对应的代理对象[Object类型，需要强转为当前代码所在对象的类型来调用所在对象的b方法或者c方法]，直接通过代理对象来调用b方法或者c方法，这样b方法和c方法的事务注解包括此前的缓存注解才会生效，示例方法如下所示：

         xxxxxxxxxx
         @Transactional(timeout=30)
         public void a(){
             OrderServiceImpl orderService = (OrderServiceImpl)AopContext.currentProxy();
             orderService.b();
             orderService.c();
         }
         @Transactional(propagation=Propagation.REQUIRED,timeout=2)
         public void b(){
             
         }
         @Transactional(propagation=Propagation.REQUIRES_NEW,timeout=20)
         public void b(){
             
         }
         

4. 分布式环境下本地事务存在问题

   • 虽然我们给创建订单和锁定库存都分别添加了@Transactional本地事务注解来各自开启单体事务，但是因为订单服务和库存服务处在不同的服务实例，因此事务不能跨服务生效，订单成功创建但是库存没有成功扣减，只会回滚库存成功锁定商品的记录，无法回滚已经创建订单记录，这是单体事务的局限

     • 解决办法，我们直接根据远程调用的结果判断，在创建订单记录的服务中判断锁库存的状态，如果锁库存失败，我们直接在订单服务抛异常来让订单服务的事务进行回滚，这样也能控住分布式事务，让多个数据库一起回滚

   • 但是通过抛出异常和单体事务结合的方案不能完美解决事务问题

     • 被调用服务成功执行，调用服务可能由于网络中断、调用超时导致被调用服务成功执行，调用服务回滚[这种情况叫远程服务假失败]。因为被调用服务本身可能不会出现异常正常执行，但是由于网络中断，系统卡死导致响应超时等都会导致在调用方抛出异常，这样就会导致远程调用成功添加记录，但是调用方由于网络，远程调用超时而抛出异常来回滚，导致订单相关数据没有一起回滚；比如锁定库存服务响应慢，库存锁定成功了，但是订单服务远程调用超时，订单服务感知到远程调用出问题了抛异常事务回滚，但是远程服务正常执行不会进行回滚，就会出现订单取消了，但是库存给锁定了

     • 已经被调用并成功执行的远程服务在订单创建失败的情况下无法自动进行回滚。远程服务调用期间没有出现问题，方法执行结束事务就已经结束了，后续订单服务运行期间出现任何问题导致需要回滚，已经执行完毕的远程服务无法自动进行事务回滚，我们想要手动回滚还需要专门写释放对应库存记录的方法

   • 本地事务在分布式系统下只能控制住本地连接的事务回滚，控制不了其他服务和连接的事务回滚，在分布式系统下，本地事务控制不住事务的根本原因是网络中断+不同数据库+服务实例集群，而本地事务只能控制一个连接内的事务